package com.atguigu.networkflow_analysis

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.AllWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.collection.immutable.ListSet

/**
 * 网站独立访客数（UV）的统计
 * 日活
 *
 * Project: UserBehaviorAnalysis
 * Package: com.atguigu.networkflow_analysis
 * Version: 1.0
 *
 * Created by  WangJX  on 2019/12/12 9:25
 */
object UniqueVisitor {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)


    val path: String = UniqueVisitor.getClass.getClassLoader.getResource("UserBehavior.csv").getPath

    val countData: DataStream[UvCount] = env.readTextFile(path)
      .map(
        data => {
          val dataArrays: Array[String] = data.split(",")
          UserBehavior(dataArrays(0).trim.toLong, dataArrays(1).trim.toLong, dataArrays(2).trim.toInt, dataArrays(3).trim, dataArrays(4).trim.toLong)
        }
      )
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[UserBehavior](Time.seconds(5)) {
        override def extractTimestamp(element: UserBehavior): Long = {
          element.timestamp * 1000L
        }
      })
      .filter(_.behavior == "pv")
      .timeWindowAll(Time.hours(1))
      .apply(new MyAllWindow())

    countData.print().setParallelism(1)

    env.execute("UniqueVisitor job")
  }

}

case class UserBehavior(
                         userId: Long,
                         itemId: Long,
                         categoryId: Int,
                         behavior: String,
                         timestamp: Long
                       )

case class UvCount(windowEnd: Long, count: Long)

class MyAllWindow() extends AllWindowFunction[UserBehavior, UvCount, TimeWindow]{

  //把开的全量窗口中的1小时内的全部数据进行set去重
  override def apply(window: TimeWindow, input: Iterable[UserBehavior], out: Collector[UvCount]): Unit = {
    var set = new ListSet[Long]()

    for (elem <- input) {
      set += elem.userId
    }

    out.collect(UvCount(window.getEnd, set.size))
  }
}
